package com.gadgets.framework.thread.pool;

import com.alibaba.fastjson.JSON;
import com.gadgets.framework.common.constant.GadgetsExceptionConstant;
import com.gadgets.framework.common.constant.ThreadEnum;
import com.gadgets.framework.common.dto.Pagination;
import com.gadgets.framework.common.exception.GadgetsValidateException;
import com.gadgets.framework.common.util.StringUtils;
import com.gadgets.framework.thread.basic.ThreadPoolConfiguration;
import com.gadgets.framework.thread.event.GadgetsThreadTaskEvent;
import com.gadgets.framework.thread.exception.GadgetsTaskSubmitException;
import com.gadgets.framework.thread.exception.GadgetsThreadExceptionConstant;
import com.gadgets.framework.thread.task.AbstractCallable;
import com.gadgets.framework.thread.task.AbstractRunnable;
import com.gadgets.framework.thread.task.AbstractThreadTask;
import com.gadgets.framework.thread.task.GadgetsRunnableFuture;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;

/**
 * @author UncleXiu
 * @version V1.0.0
 * @data 2020-09-12
 * 
 */
@SuppressWarnings({"rawtypes", "unchecked"})
public class GadgetsThreadPool extends ThreadPoolExecutor {

    private static final Logger logger = LoggerFactory.getLogger(GadgetsThreadPool.class);

    /**
     * 线程池状态
     * 
     *  {@link ThreadEnum#INIT}<br>
     *  {@link ThreadEnum#RUNNING}<br>
     *  {@link ThreadEnum#SUSPED}<br>
     *  {@link ThreadEnum#DESTORY}
     */
    private int poolStatus = ThreadEnum.INIT.getStatus();
    /**
     * 限流队列
     * K - V: 模块编号 - 模块限流队列
     */
    private static final ConcurrentHashMap<String,ArrayBlockingQueue<GadgetsRunnableFuture>> FLOW_LIMIT_CONTAINER = new ConcurrentHashMap<>();
    /**
     * 模块线程计数器
     * K - V: 模块编号 - 模块线程计数器
     */
    private static final ConcurrentHashMap<String,Integer> THREAD_CALCULATE_CONTAINER = new ConcurrentHashMap<>();
    /**
     * 正在执行的任务队列
     * K - V: 模块编号 - 正在执行的任务队列
     */
    private static final ConcurrentHashMap<String, ArrayList<GadgetsRunnableFuture>> RUNNING_LIMIT_CONTAINER = new ConcurrentHashMap<>();
    /**
     * 已完成的任务队列
     * K - V: 模块编号 - 已完成的任务队列
     */
    private static final ConcurrentHashMap<String,ArrayList<GadgetsRunnableFuture>> FINISHED_LIMIT_CONTAINER = new ConcurrentHashMap<>();
    /**
     * 模块配置信息
     * K - V: 模块编号 - 配置信息
     */
    private static final ConcurrentHashMap<String, ThreadPoolConfiguration> CONFIGURATION_CONTAINER = new ConcurrentHashMap<>();
    /**
     * 线程事件信息
     */
    private static final LinkedBlockingQueue<GadgetsThreadTaskEvent> EVENT_ARRAY_LIST = new LinkedBlockingQueue<>();
    /**
     * 非实时线程事件信息（查询用，不可用于逻辑处理）
     */
    private static Object[] EVENT_UNACTUAL_QUERY_ARRAY = null;
    /**
     * 非实时限流队列事件查询（查询用，不可用于逻辑处理）
     */
    private static ConcurrentHashMap<String ,Object[]> LIMIT_FLOW_UNACTUAL_QUERY_ARRAY = new ConcurrentHashMap<>();
    /**
     * 任务调度线程休眠时间
     */
    private static Integer sleepMilSecond = 200;
    /**
     * 线程池 升级/降级 阈值
     */
    private static Integer levelUpThreshold = 0;
    private static Integer levelDownThreshold = 0;

    /**
     * Creates a new {@code ThreadPoolExecutor} with the given initial
     * parameters and default thread factory and rejected execution handler.
     * It may be more convenient to use one of the {@link Executors} factory
     * methods instead of this general purpose constructor.
     *
     * @param corePoolSize    the number of threads to keep in the pool, even
     *                        if they are idle, unless {@code allowCoreThreadTimeOut} is set
     * @param maximumPoolSize the maximum number of threads to allow in the
     *                        pool
     * @param keepAliveTime   when the number of threads is greater than
     *                        the core, this is the maximum time that excess idle threads
     *                        will wait for new tasks before terminating.
     * @param unit            the time unit for the {@code keepAliveTime} argument
     * @param workQueue       the queue to use for holding tasks before they are
     *                        executed.  This queue will hold only the {@code Runnable}
     *                        tasks submitted by the {@code execute} method.
     *
     * @throws IllegalArgumentException if one of the following holds:<br>
     *                                  {@code corePoolSize < 0}<br>
     *                                  {@code keepAliveTime < 0}<br>
     *                                  {@code maximumPoolSize <= 0}<br>
     *                                  {@code maximumPoolSize < corePoolSize}
     * @throws NullPointerException     if {@code workQueue} is null
     */
    public GadgetsThreadPool(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, Integer levelUp, Integer levelDown) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue);
        if(corePoolSize<4){
            throw new GadgetsValidateException(GadgetsThreadExceptionConstant.THREAD_POOL_ERROR,"核心线程数不足");
        }
        levelUpThreshold = levelUp;
        levelDownThreshold = levelDown;
        poolStatus = ThreadEnum.RUNNING.getStatus();
        FlowMonitorRunnable flowMonitorRunnable = new FlowMonitorRunnable();
        super.execute(flowMonitorRunnable);
        TaskDispatchRunnable taskDispatchRunnable = new TaskDispatchRunnable();
        super.execute(taskDispatchRunnable);
        HookDispatchRunnable hookDispatchRunnable = new HookDispatchRunnable();
        super.execute(hookDispatchRunnable);
    }

    /**
     * Creates a new {@code ThreadPoolExecutor} with the given initial
     * parameters and default rejected execution handler.
     *
     * @param corePoolSize    the number of threads to keep in the pool, even
     *                        if they are idle, unless {@code allowCoreThreadTimeOut} is set
     * @param maximumPoolSize the maximum number of threads to allow in the
     *                        pool
     * @param keepAliveTime   when the number of threads is greater than
     *                        the core, this is the maximum time that excess idle threads
     *                        will wait for new tasks before terminating.
     * @param unit            the time unit for the {@code keepAliveTime} argument
     * @param workQueue       the queue to use for holding tasks before they are
     *                        executed.  This queue will hold only the {@code Runnable}
     *                        tasks submitted by the {@code execute} method.
     * @param threadFactory   the factory to use when the executor
     *                        creates a new thread
     *
     * @throws IllegalArgumentException if one of the following holds:<br>
     *                                  {@code corePoolSize < 0}<br>
     *                                  {@code keepAliveTime < 0}<br>
     *                                  {@code maximumPoolSize <= 0}<br>
     *                                  {@code maximumPoolSize < corePoolSize}
     * @throws NullPointerException     if {@code workQueue}
     *                                  or {@code threadFactory} is null
     */
    public GadgetsThreadPool(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, Integer levelUp, Integer levelDown) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory);
        if(corePoolSize<4){
            throw new GadgetsValidateException(GadgetsThreadExceptionConstant.THREAD_POOL_ERROR,"核心线程数不足");
        }
        levelUpThreshold = levelUp;
        levelDownThreshold = levelDown;
        poolStatus = ThreadEnum.RUNNING.getStatus();
        FlowMonitorRunnable flowMonitorRunnable = new FlowMonitorRunnable();
        super.execute(flowMonitorRunnable);
        TaskDispatchRunnable taskDispatchRunnable = new TaskDispatchRunnable();
        super.execute(taskDispatchRunnable);
        HookDispatchRunnable hookDispatchRunnable = new HookDispatchRunnable();
        super.execute(hookDispatchRunnable);
    }

    /**
     * Creates a new {@code ThreadPoolExecutor} with the given initial
     * parameters and default thread factory.
     *
     * @param corePoolSize    the number of threads to keep in the pool, even
     *                        if they are idle, unless {@code allowCoreThreadTimeOut} is set
     * @param maximumPoolSize the maximum number of threads to allow in the
     *                        pool
     * @param keepAliveTime   when the number of threads is greater than
     *                        the core, this is the maximum time that excess idle threads
     *                        will wait for new tasks before terminating.
     * @param unit            the time unit for the {@code keepAliveTime} argument
     * @param workQueue       the queue to use for holding tasks before they are
     *                        executed.  This queue will hold only the {@code Runnable}
     *                        tasks submitted by the {@code execute} method.
     * @param handler         the handler to use when execution is blocked
     *                        because the thread bounds and queue capacities are reached
     *
     * @throws IllegalArgumentException if one of the following holds:<br>
     *                                  {@code corePoolSize < 0}<br>
     *                                  {@code keepAliveTime < 0}<br>
     *                                  {@code maximumPoolSize <= 0}<br>
     *                                  {@code maximumPoolSize < corePoolSize}
     * @throws NullPointerException     if {@code workQueue}
     *                                  or {@code handler} is null
     */
    public GadgetsThreadPool(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, RejectedExecutionHandler handler, Integer levelUp, Integer levelDown) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, handler);
        if(corePoolSize<4){
            throw new GadgetsValidateException(GadgetsThreadExceptionConstant.THREAD_POOL_ERROR,"核心线程数不足");
        }
        levelUpThreshold = levelUp;
        levelDownThreshold = levelDown;
        poolStatus = ThreadEnum.RUNNING.getStatus();
        FlowMonitorRunnable flowMonitorRunnable = new FlowMonitorRunnable();
        super.execute(flowMonitorRunnable);
        TaskDispatchRunnable taskDispatchRunnable = new TaskDispatchRunnable();
        super.execute(taskDispatchRunnable);
        HookDispatchRunnable hookDispatchRunnable = new HookDispatchRunnable();
        super.execute(hookDispatchRunnable);
    }

    /**
     * Creates a new {@code ThreadPoolExecutor} with the given initial
     * parameters.
     *
     * @param corePoolSize    the number of threads to keep in the pool, even
     *                        if they are idle, unless {@code allowCoreThreadTimeOut} is set
     * @param maximumPoolSize the maximum number of threads to allow in the
     *                        pool
     * @param keepAliveTime   when the number of threads is greater than
     *                        the core, this is the maximum time that excess idle threads
     *                        will wait for new tasks before terminating.
     * @param unit            the time unit for the {@code keepAliveTime} argument
     * @param workQueue       the queue to use for holding tasks before they are
     *                        executed.  This queue will hold only the {@code Runnable}
     *                        tasks submitted by the {@code execute} method.
     * @param threadFactory   the factory to use when the executor
     *                        creates a new thread
     * @param handler         the handler to use when execution is blocked
     *                        because the thread bounds and queue capacities are reached
     *
     * @throws IllegalArgumentException if one of the following holds:<br>
     *                                  {@code corePoolSize < 0}<br>
     *                                  {@code keepAliveTime < 0}<br>
     *                                  {@code maximumPoolSize <= 0}<br>
     *                                  {@code maximumPoolSize < corePoolSize}
     * @throws NullPointerException     if {@code workQueue}
     *                                  or {@code threadFactory} or {@code handler} is null
     */
    public GadgetsThreadPool(int corePoolSize, int maximumPoolSize, long keepAliveTime, TimeUnit unit, BlockingQueue<Runnable> workQueue, ThreadFactory threadFactory, RejectedExecutionHandler handler, Integer levelUp, Integer levelDown) {
        super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory, handler);
        if(corePoolSize<4){
            throw new GadgetsValidateException(GadgetsThreadExceptionConstant.THREAD_POOL_ERROR,"核心线程数不足");
        }
        levelUpThreshold = levelUp;
        levelDownThreshold = levelDown;
        poolStatus = ThreadEnum.RUNNING.getStatus();
        FlowMonitorRunnable flowMonitorRunnable = new FlowMonitorRunnable();
        super.execute(flowMonitorRunnable);
        TaskDispatchRunnable taskDispatchRunnable = new TaskDispatchRunnable();
        super.execute(taskDispatchRunnable);
        HookDispatchRunnable hookDispatchRunnable = new HookDispatchRunnable();
        super.execute(hookDispatchRunnable);
    }

    /**
     * Initiates an orderly shutdown in which previously submitted
     * tasks are executed, but no new tasks will be accepted.
     * Invocation has no additional effect if already shut down.
     *
     * <p>This method does not wait for previously submitted tasks to
     * complete execution.  Use {@link #awaitTermination awaitTermination}
     * to do that.
     *
     * @throws SecurityException {@inheritDoc}
     */
    @Override
    public void shutdown() {
        poolStatus = ThreadEnum.DESTORY.getStatus();
        super.shutdown();
    }

    /**
     * Attempts to stop all actively executing tasks, halts the
     * processing of waiting tasks, and returns a list of the tasks
     * that were awaiting execution. These tasks are drained (removed)
     * from the task queue upon return from this method.
     *
     * <p>This method does not wait for actively executing tasks to
     * terminate.  Use {@link #awaitTermination awaitTermination} to
     * do that.
     *
     * <p>There are no guarantees beyond best-effort attempts to stop
     * processing actively executing tasks.  This implementation
     * cancels tasks via {@link Thread#interrupt}, so any task that
     * fails to respond to interrupts may never terminate.
     *
     * @throws SecurityException {@inheritDoc}
     */
    @Override
    public List<Runnable> shutdownNow() {
        poolStatus = ThreadEnum.DESTORY.getStatus();
        return super.shutdownNow();
    }

    /**
     * 线程任务提交校验
     * @param task              线程任务
     * @return                  线程任务
     * @throws RuntimeException 非捕获异常
     */
    protected AbstractThreadTask taskTypeValidate(Object task) throws RuntimeException{
        if(poolStatus!=ThreadEnum.RUNNING.getStatus()){
            // 线程池状态如果不是执行状态，则不允许提交任务
            throw new GadgetsValidateException(GadgetsExceptionConstant.VALIDATE_ERROR, "线程池状态不是执行中，已不接受新任务提交");
        }
        if(null==task){
            // 任务对象不能为空指针
            throw new NullPointerException("Task can't be null");
        } else if(task instanceof AbstractThreadTask) {
            // 如果任务类型是 AbstractThreadTask 就继续提交
            return (AbstractThreadTask)task;
        } else {
            // 任务类型不正确，抛出异常信息
            throw new GadgetsValidateException(GadgetsExceptionConstant.VALIDATE_ERROR, "任务类型不正确");
        }
    }

    /**
     * 模块注册
     * @param configuration 配置文件
     */
    public static void registryModuel(ThreadPoolConfiguration configuration){
        logger.info("模块{}发起注册，注册信息{}",configuration.getModuelNo(),JSON.toJSONString(configuration));
        FLOW_LIMIT_CONTAINER.put(configuration.getModuelNo(), new ArrayBlockingQueue<GadgetsRunnableFuture>(configuration.getFlowLimitSize()));
        THREAD_CALCULATE_CONTAINER.put(configuration.getModuelNo(),configuration.getMinThread());
        RUNNING_LIMIT_CONTAINER.put(configuration.getModuelNo(),new ArrayList<>(configuration.getMaxThread()));
        FINISHED_LIMIT_CONTAINER.put(configuration.getModuelNo(),new ArrayList<>((int)(0.4*configuration.getFlowLimitSize())));
        CONFIGURATION_CONTAINER.put(configuration.getModuelNo(),configuration);
        logger.info("模块{}注册完成",configuration.getModuelNo());
    }

    /**
     * 提交任务
     * @param task  线程任务
     * @return      任务钩子
     */
    @Override
    public Future<?> submit(Runnable task) {
        return submit(task,null);
    }

    /**
     * 提交任务
     * @param task      线程任务
     * @param result    任务钩子
     *
     * @throws RejectedExecutionException {@inheritDoc}
     * @throws NullPointerException       {@inheritDoc}
     */
    @Override
    public <T> Future<T> submit(Runnable task, T result) {
        AbstractThreadTask threadTask = taskTypeValidate(task);
        GadgetsRunnableFuture<T> ftask = (GadgetsRunnableFuture<T>) newTaskFor(task, result);
        submit(threadTask, ftask);
        return ftask;
    }

    /**
     * 提交任务
     * @param task      线程任务
     *
     * @throws RejectedExecutionException {@inheritDoc}
     * @throws NullPointerException       {@inheritDoc}
     */
    @Override
    public <T> Future<T> submit(Callable<T> task) {
        AbstractThreadTask threadTask = taskTypeValidate(task);
        GadgetsRunnableFuture<T> ftask = (GadgetsRunnableFuture<T>) newTaskFor(task);
        submit(threadTask, ftask);
        return ftask;
    }

    /**
     * 线程任务提交线程池
     * @param threadTask    线程任务
     * @param ftask         任务钩子
     * @param <T>           任务类型
     */
    protected <T> void submit(AbstractThreadTask threadTask, GadgetsRunnableFuture<T> ftask) {
        String moduelNo = threadTask.getModuelNo();
        if(StringUtils.isBlank(moduelNo)){
            throw new GadgetsValidateException(GadgetsExceptionConstant.VALIDATE_ERROR,"模块号不能为空");
        }
        if(!FLOW_LIMIT_CONTAINER.containsKey(moduelNo) || null==FLOW_LIMIT_CONTAINER.get(moduelNo)){
            // 如果无法获取限流队列，抛出异常
            throw new GadgetsValidateException(GadgetsThreadExceptionConstant.MODUEL_NOT_EXIST,moduelNo);
        } else {
            // 将任务加入到限流队列中
            ArrayBlockingQueue<GadgetsRunnableFuture> blockingQueue = FLOW_LIMIT_CONTAINER.get(moduelNo);
            try {
                blockingQueue.add(ftask);
            } catch (Throwable e) {
                // 限流队列已满
                throw new GadgetsTaskSubmitException(GadgetsThreadExceptionConstant.MODUEL_FLOW_LIMIT_END,moduelNo);
            }
        }
    }

    /**
     * 执行方法
     * @param command   线程任务
     */
    @Override
    public void execute(Runnable command) {
        AbstractThreadTask threadTask = taskTypeValidate(command);
        GadgetsRunnableFuture ftask = (GadgetsRunnableFuture) newTaskFor(command,null);
        submit(threadTask, ftask);
    }


    @Override
    protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) {
        AbstractRunnable abstractRunnable = (AbstractRunnable) runnable;
        GadgetsRunnableFuture<T> runnableFuture = new GadgetsRunnableFuture<T>(runnable, value);
        runnableFuture.setTask(abstractRunnable);
        return runnableFuture;
    }


    @Override
    protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
        AbstractCallable<T> abstractCallable = (AbstractCallable<T>) callable;
        GadgetsRunnableFuture<T> runnableFuture = new GadgetsRunnableFuture(callable);
        runnableFuture.setTask(abstractCallable);
        return runnableFuture;
    }

    /**
     * 线程池发布执行事件
     * @param event 事件类型
     */
    public void publishEvent(GadgetsThreadTaskEvent event){
        EVENT_ARRAY_LIST.add(event);
    }

    /**
     * 线程流量监听线程
     */
    protected final class FlowMonitorRunnable extends AbstractRunnable {
        @Override
        public void run() {
            while (ThreadEnum.RUNNING.getStatus() == poolStatus) {
                try {
                    Thread.sleep(10 * 1000);
                } catch (InterruptedException e) {
                    logger.error("流量控制器休眠失败");
                }
                if (ThreadEnum.RUNNING.getStatus() == poolStatus) {
                    logger.debug("流量控制器启动");
                    for (Map.Entry<String, ThreadPoolConfiguration> configurationEntry : CONFIGURATION_CONTAINER.entrySet()) {
                        String moduelNo = configurationEntry.getKey();
                        ThreadPoolConfiguration configuration = configurationEntry.getValue();
                        // 获取线程队列信息
                        ArrayBlockingQueue<GadgetsRunnableFuture> link = FLOW_LIMIT_CONTAINER.get(moduelNo);
                        if (link.size() > configuration.getLevelUpNum() && configuration.getMaxThread() > configuration.getCurrentThread()) {
                            logger.info("模块{}线程升级", moduelNo);
                            // 如果当前模块积累任务已经超过升级阈值，且当前模块没有到最大线程数，就线程数+1
                            configuration.levelUp();
                            synchronized (THREAD_CALCULATE_CONTAINER) {
                                THREAD_CALCULATE_CONTAINER.put(moduelNo, Math.min(configuration.getCurrentThread(),THREAD_CALCULATE_CONTAINER.get(moduelNo) + 1));
                            }
                        } else if (link.size() < configuration.getLevelDownNum() && configuration.getMinThread() < configuration.getCurrentThread()) {
                            logger.info("模块{}线程降级", moduelNo);
                            // 如果当前模块积累任务已经低于降级阈值，且当前模块没有到最小线程数，就线程数-1
                            configuration.levelDown();
                            synchronized (THREAD_CALCULATE_CONTAINER) {
                                THREAD_CALCULATE_CONTAINER.put(moduelNo, Math.min(configuration.getCurrentThread(),THREAD_CALCULATE_CONTAINER.get(moduelNo) + 1));
                            }
                        }
                    }
                } else {
                    logger.debug("线程池未启动，流量控制器跳过执行");
                }
            }
        }
    }

    /**
     * 任务调度线程
     */
    protected final class TaskDispatchRunnable extends AbstractRunnable {
        @Override
        public void run() {
            try {
                // 等待任务更新计数器，数据更新变量，默认不更新
                int num = 0;
                boolean flag = false;
                while (ThreadEnum.RUNNING.getStatus() == poolStatus) {
                    try {
                        Thread.sleep(sleepMilSecond);
                    } catch (InterruptedException e) {
                        logger.error("任务调度器休眠失败");
                    }
                    int taskNum = 0;
                    num++;
                    flag = (0==num%40);
                    for (Map.Entry<String, ThreadPoolConfiguration> configurationEntry : CONFIGURATION_CONTAINER.entrySet()) {
                        String moduelNo = configurationEntry.getKey();
                        logger.trace("任务调度器当前处理模块{}", moduelNo);
                        ArrayBlockingQueue<GadgetsRunnableFuture> link = FLOW_LIMIT_CONTAINER.get(moduelNo);
                        ArrayList<GadgetsRunnableFuture> runningLink = RUNNING_LIMIT_CONTAINER.get(moduelNo);
                        ArrayList<GadgetsRunnableFuture> finishLink = FINISHED_LIMIT_CONTAINER.get(moduelNo);
                        Integer calculateNum = 0;
                        // 计算总任务数
                        taskNum += link.size();
                        // 移除已完成任务
                        romoveFinishTask(moduelNo, runningLink, finishLink, calculateNum);
                        // 从任务队列中 获取队列第一个任务加入 进行中任务
                        addRunningTask(moduelNo, link, runningLink, calculateNum);
                        // 计数器符合条件时，更新数据快照
                        if(flag){
                            LIMIT_FLOW_UNACTUAL_QUERY_ARRAY.put(moduelNo,link.toArray());
                        }
                    }
                    // 判断是否需要线程调度级别调整
                    if (taskNum < levelDownThreshold && sleepMilSecond > 50) {
                        // 调度降级
                        sleepMilSecond -= 50;
                    } else if (taskNum > levelUpThreshold && sleepMilSecond < 800) {
                        // 调度升级
                        sleepMilSecond += 50;
                    }
                }
            } catch (Throwable e) {
                logger.error("线程调度发生异常："+JSON.toJSONString(e));
            }
        }

        /**
         * 调度需要执行任务
         * @param moduelNo      模块号
         * @param link          待处理任务
         * @param runningLink   运行中任务
         * @param calculateNum  线程计数
         */
        protected void addRunningTask(String moduelNo, ArrayBlockingQueue<GadgetsRunnableFuture> link, ArrayList<GadgetsRunnableFuture> runningLink, Integer calculateNum) {
            GadgetsRunnableFuture ftask = link.poll();
            if(null != ftask){
                // 如果存在线程任务，提交执行
                synchronized (THREAD_CALCULATE_CONTAINER) {
                    THREAD_CALCULATE_CONTAINER.put(moduelNo, THREAD_CALCULATE_CONTAINER.get(moduelNo) - 1);
                }
                ThreadPoolConfiguration configuration = CONFIGURATION_CONTAINER.get(moduelNo);
                runningLink.add(ftask);
                try {
                    GadgetsThreadPool.super.execute(ftask);
                } catch (Throwable e) {
                    logger.error("线程任务{}提交失败", JSON.toJSONString(ftask));
                    logger.error(StringUtils.formatExceptionStack(e));
                    runningLink.remove(ftask);
                    synchronized (THREAD_CALCULATE_CONTAINER) {
                        THREAD_CALCULATE_CONTAINER.put(moduelNo, Math.min(configuration.getCurrentThread(),THREAD_CALCULATE_CONTAINER.get(moduelNo) + 1));
                    }
                }
            }else {
                logger.trace("模块{}没有需要执行的任务",moduelNo);
            }
        }

        /**
         * 移除已完成任务
         * @param moduelNo      模块编号
         * @param runningLink   运行中任务
         * @param finishLink    已完成任务
         * @param calculateNum  线程计数
         */
        protected void romoveFinishTask(String moduelNo, ArrayList<GadgetsRunnableFuture> runningLink, ArrayList<GadgetsRunnableFuture> finishLink, Integer calculateNum) {
            // 从进行中任务 移除已完成任务 到 已完成队列
            ThreadPoolConfiguration configuration = CONFIGURATION_CONTAINER.get(moduelNo);
            Iterator<GadgetsRunnableFuture> iterator = runningLink.iterator();
            while (iterator.hasNext()){
                GadgetsRunnableFuture future = iterator.next();
                if(future.isDone()){
                    synchronized (THREAD_CALCULATE_CONTAINER) {
                        THREAD_CALCULATE_CONTAINER.put(moduelNo, Math.min(configuration.getCurrentThread(),THREAD_CALCULATE_CONTAINER.get(moduelNo) + 1));
                    }
                    iterator.remove();
                    synchronized (FINISHED_LIMIT_CONTAINER.get(moduelNo)){
                        finishLink.add(future);
                    }
                }
            }
        }
    }

    /**
     * 线程事件钩子调度器
     */
    protected final class HookDispatchRunnable extends AbstractRunnable{
        @Override
        public void run() {
            // 更新查询队列计数器
            int num = 0;
            while (ThreadEnum.RUNNING.getStatus() == poolStatus) {
                try {
                    Thread.sleep(sleepMilSecond);
                } catch (InterruptedException e) {
                    logger.error("线程钩子调度器休眠失败");
                }
                // 处理已完成任务调度器
                for (String moduelNo : CONFIGURATION_CONTAINER.keySet()) {
                    synchronized (FINISHED_LIMIT_CONTAINER.get(moduelNo)) {
                        Iterator<GadgetsRunnableFuture> iterator = FINISHED_LIMIT_CONTAINER.get(moduelNo).iterator();
                        while (iterator.hasNext()) {
                            GadgetsRunnableFuture finishedTask = iterator.next();
                            iterator.remove();
                        }
                    }
                }
                GadgetsThreadTaskEvent event;
                // 处理事件钩子
                int i = EVENT_ARRAY_LIST.size();
                while (i != 0) {
                    event = EVENT_ARRAY_LIST.poll();
                    if (null != event) {
                        if (event.excuteHook()) {
                            logger.info("任务{}的钩子{}执行成功", event.getTaskNo(), event.getClass().getName());
                        } else {
                            logger.error("任务{}的钩子{}执行失败", event.getTaskNo(), event.getClass().getName());
                        }
                    }
                    i--;
                }
                // 更新队列计数器+1，每40次轮询更新一次事件队列，并重置计数器
                num++;
                if(num%40==0){
                    // 从事件主线中获取事件信息列表
                    EVENT_UNACTUAL_QUERY_ARRAY = EVENT_ARRAY_LIST.toArray();
                    num = 0;
                }
            }
        }
    }

    public int getPoolStatus() {
        return poolStatus;
    }

    public void setPoolStatus(int poolStatus) {
        this.poolStatus = poolStatus;
    }

    /**
     * 分页获取事件主线快照
     * @param pagination    分页条件
     * @return              查询结果
     */
    public static List<Object> getEventQueryArray(Pagination pagination){
        List<Object> list = new ArrayList<>();
        int startIndex = pagination.getPageSize()*pagination.getCurrentPage();
        int endIndex = pagination.getPageSize()*(pagination.getCurrentPage()+1);
        while (startIndex<endIndex){
            list.add(EVENT_UNACTUAL_QUERY_ARRAY[startIndex]);
            startIndex++;
        }
        return list;
    }

    /**
     * 分页获取模块限流队列快照
     * @param moduelNo      模块号
     * @param pagination    分页条件
     * @return              查询结果
     */
    public static List<Object> getLimitFlowQueryArray(String moduelNo, Pagination pagination){
        List<Object> list = new ArrayList<>();
        int startIndex = pagination.getPageSize()*pagination.getCurrentPage();
        int endIndex = pagination.getPageSize()*(pagination.getCurrentPage()+1);
        if(null!=LIMIT_FLOW_UNACTUAL_QUERY_ARRAY.get(moduelNo)) {
            while (startIndex < endIndex) {
                list.add(LIMIT_FLOW_UNACTUAL_QUERY_ARRAY.get(moduelNo)[startIndex]);
                startIndex++;
            }
        }
        return list;
    }

    /**
     * 获取模块配置信息
     * @return     查询结果
     */
    public static List<ThreadPoolConfiguration> getModuelConfiguration(){
        List<ThreadPoolConfiguration> list = new ArrayList<>();
        for (ThreadPoolConfiguration value : CONFIGURATION_CONTAINER.values()) {
            list.add(value);
        }
        return list;
    }

    /**
     * 重置模块程序计数器
     * @param moduelNo  模块号
     */
    public static void resetThreadCalculate(String moduelNo){
        synchronized (THREAD_CALCULATE_CONTAINER) {
            ThreadPoolConfiguration configuration = CONFIGURATION_CONTAINER.get(moduelNo);
            THREAD_CALCULATE_CONTAINER.put(moduelNo, configuration.getCurrentThread());
        }
    }

}
